package org.example

import org.apache.spark.api.java.JavaRDD.fromRDD
import org.apache.spark.sql.SparkSession

import java.util.regex.Pattern

object movies {
   def main(args:Array[String]): Unit = {
     val spark = SparkSession
       .builder()
       .master("local[*]")
       .appName("spark")
       .getOrCreate()
     val sc = spark.sparkContext
     //电影用户的分析数据概况
     val filePath = "src/main/resources/"
     val moviesRDD = sc.textFile(filePath + "movies.dat")
     val occupationsRDD = sc.textFile(filePath + "occupations.dat")
     val ratingsRDD = sc.textFile(filePath + "ratings.dat")
     val usersRDD = sc.textFile(filePath + "users.dat")
     println("总电影数：" + moviesRDD.count())
     println("总用户数：" + usersRDD.count())
     println("总评分数：" + ratingsRDD.count())
     println("职业数：" + occupationsRDD.count())

     val userRating = ratingsRDD.map(_.split("::"))
       .map(user => (user(1),user(0)))
       .filter(_._2.equals("2688"))
     println("用户为2688统计：" + usersRDD.count())

     val movieInfo = moviesRDD.map(_.split("::"))
       .map(movie => (movie(0),(movie(1),movie(2))))
       .join(userRating)
       .map(item => (item._1, item._2._2,item._2._1._2))

     movieInfo.take(5).foreach(println)


     val rating = ratingsRDD.map(x => x.split("::")).map {
       x => {
         (x(0), x(1), x(2)) // (UserID,MovieID,Rating)
          }
        }.cache()
       println("平均得分最高的前 10 名的电影名称简单版")
       rating.map(x => (x._2, (x._3.toDouble, 1)))
     //  ( MovieId,( 总分, 总次数) )
        .reduceByKey((x, y) => {
          (x._1 + y._1, x._2 + y._2)
     }) // (  平均分, MovieId )
           .map(x => (x._2._1 / x._2._2, x._1))
           .sortByKey(ascending = false)
           .take(5)
           .foreach(println)

         println("按平均分取前 10 部电影输出详情:(平均分,(movieId,Title,Genres,总分,总次 数))")
          // MovieID::Title::Genres \
         val moviesInfo = moviesRDD.map(x => x.split("::"))
           .map(x => {
             (x(0), (x(1), x(2)))
           })

         val ratingsInfo = rating.map(x => (x._2, (x._3.toDouble, 1)))
     //  ( MovieId,( Rating, 1) )  (1252,(4.0,1))
           .reduceByKey((x, y) => {
       (x._1 + y._1, x._2 + y._2)
       }) //( MovieId,( 总分, 总次数) )
        .map(x => (x._1, (x._2._1 / x._2._2, x._2._1, x._2._2)))
     //  ( MovieId, (平均分, 总分,总次数) )
         moviesInfo.join(ratingsInfo)
         .map(info => {
           (info._2._2._1, (info._1, info._2._1._1, info._2._1._2, info._2._2._2, info._2._2._3))
            // (平均分,(movieId,Title,Genres,总分,总次数))
         })  .sortByKey(false)
             .take(5)
             .foreach(println)
     println("观影人数最多的前 10 部电影")
     // (UserID,MovieID,Rating)
     val watchViewsInfo = rating.map(x => {
         (x._2, 1)
       }).reduceByKey((x, y) => x + y) // ( MovieId,总次数 )
       .map(x => (x._2, x._1))
       .sortByKey(false)
       .take(5)
     // 5 名
     watchViewsInfo.foreach(println(_))
     println("===================>")
     rating.map(x => (x._2, 1)) // ( MovieId, 1)
       .reduceByKey((x, y) => {
         (x + y)
       }) // ( MovieId,总次数 )
       .map(x => (x._2, x._1)) // ( 总次数, MovieId )
       .sortByKey(false)
       .take(10)
       .foreach(println) // 286-> 999
     println("详情的输出( 观影人数，电影编号)")
     // 输出 ( 总次数, (MovieID, title,Genres,总分,平均分 ) )
     // ratingsInfo ( MovieId, (平均分, 总分,总次数) )
     // ( MovieID, ( (Title,Genres),(平均分, 总分,总次数) ) )
     // ratingsInfo ( MovieId, (平均分, 总分,总次数) )
     // moviesInfo MovieID::Title::Genres
     moviesInfo.join(ratingsInfo).map(x => {
         (x._2._2._3, (x._2._1._1, x._2._1._2, x._2._2._1, x._2._2._1))
       }).sortByKey(false)
       .take(10)
       .foreach(println)

    println("打印喜剧片（Comedy）的数量")
     val movieStyle = moviesRDD.map(_.split("::"))
       .map(tp => (tp(0), tp(2))) //只留编号和类型
       .flatMapValues(sp => sp.split("\\|"))
       .map(p => (p._2, 1)) //转换成键值对
       .reduceByKey((x, y) => x + y) //聚合
       .filter(_._1.equals("Comedy"))
         movieStyle.foreach(println)


     println("年度电影数量统计")
     val pattern = Pattern.compile("(.*) (\\(\\d{4}\\))")

     val movieYear = moviesRDD.map(_.split("::"))
       .map(x => (x(1),1))
       .map(x => {
         var name = ""
         var year = ""
         var matchre = pattern.matcher(x._1)
         if (matchre.find()) {
           name = matchre.group(1)
           year = matchre.group(2)
           year = year.substring(1,year.length - 1)
         }
         if (year == "") {
           (-1,1) //标识码
         } else {
           (year.toInt,1)
         }
       })
       .reduceByKey((x,y) => x+y)
       .filter(d => (d._1 ==2000))





       movieYear.foreach(println)
         sc.stop()
   }
}
